package my.thesis.spouts;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.util.Date;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.FilterQuery;
import twitter4j.GeoLocation;
import twitter4j.HashtagEntity;
import twitter4j.Place;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterObjectFactory;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.URLEntity;
import twitter4j.User;
import twitter4j.UserMentionEntity;
import twitter4j.conf.ConfigurationBuilder;
import twitter4j.json.DataObjectFactory;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import org.joda.time.DateTime;
import org.json.JSONArray;
import org.json.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;

public class TweetReader extends BaseRichSpout {

	private LinkedBlockingQueue<Status> queue = null;
	private TwitterStream twitterStream;
	private int queueDepth;
	private SpoutOutputCollector collector;
	private int counter=0;
	
	
	//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********
	
	File file;
	FileWriter fw;
	BufferedWriter bw;
	
	//************* SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT ********
	
	public void nextTuple() 
	{
		
		//retrieves and removes head of queue 
		Status ret = queue.poll();
		
		if(ret == null)
		{
			try 
			{ 
				Thread.sleep(50); 
			}
			catch (InterruptedException e) {}
		} 
		else 
		{	        
			collector.emit(new Values(ret));
		}
		
	}

	public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) 
	{
		queueDepth=2000;
		this.collector = collector;
		queue = new LinkedBlockingQueue<Status>(queueDepth);
		
		//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********
		
		File file = new File("MY_TWEETS_HASHTAGS_29_30_08_40.json");

		if(!file.exists())
		{
			try {
				file.createNewFile();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		
		try 
		{
			fw = new FileWriter(file.getAbsoluteFile());
		} 
		catch (IOException e1) 
		{
			
			e1.printStackTrace();
		}
		
		bw = new BufferedWriter(fw);
		
		//************ SINGLE SPOUT TOPOLOGY GIA EGGRAFH SE TEXT *********
		
		 StatusListener listener = new StatusListener()
		 {

			public void onException(Exception arg0) {
				// TODO Auto-generated method stub
				
			}

			public void onDeletionNotice(StatusDeletionNotice arg0) {
				// TODO Auto-generated method stub
				
			}

			public void onScrubGeo(long arg0, long arg1) {
				// TODO Auto-generated method stub
				
			}

			public void onStallWarning(StallWarning arg0) {
				// TODO Auto-generated method stub
				
			}

			public void onStatus(Status status) 
			{
				if(queue.size() < queueDepth)
	    		{
	    			//System.out.println("TWEET Received: " + status.getText());
	    			
	    			//inserts at tail of queue
	    			queue.offer(status);
	    			
	    		    if(status.getHashtagEntities().length == 0)
	    			{
	    				//do nothing
	    			}
	    			else
	    			{
	    				//*******************
	    				
	    				Long id = status.getId();
	    				
	    				User user =status.getUser();
	    				
	    				String username = user.getScreenName();
	    						
	    				int retweets=status.getRetweetCount();
	    				
	    				int followers=user.getFollowersCount();
	    				
	    				//Date created = status.getCreatedAt();
	    				//long millis=created.getTime();
	    				//String createdat=created.toString();
	    				
	    				//VAZW TIMESTAMP SE KATHE TWEET THN DIKH MOY TREXOYSA WRA
	    				DateTime timestamp = DateTime.now();
	    				long millis = timestamp.getMillis();
	    				
	    				String createdat=timestamp.toString();
	    				
	    				String lang = status.getLang();
	    				
	    				String text = status.getText();
	    				
	    				GeoLocation geo = status.getGeoLocation();
	    				
	    				double lon = 0;
	    				double lat = 0;
	    				
	    				if(geo!=null)
	    				{
	    					lat=geo.getLatitude();
	    					lon=geo.getLongitude();
	    				}
	    				else
	    				{
	    					lat=Math.random()*-90 + Math.random()*90;
	    					lon=Math.random()*-180+Math.random()*180;    					
	    				}
	    							
	    				Place pla = status.getPlace();
	    				
	    				String country = "";
	    				
	    				if(pla!=null)
	    				{
	    					if(pla.getPlaceType()!=null)
	    					{				
	    						country=pla.getCountry();
	    					}
	    				}
	    				
	    				if(!(lang.equals("pt")||lang.equals("gr")||lang.equals("en")||lang.equals("en-gb")||lang.equals("fr")||lang.equals("de")||lang.equals("es")||lang.equals("it"))||(lat==0||lon==0)||country.equals("")||(status.isRetweet()==true))
	    				{
	    					//do nothing
	    				}
	    				else
	    				{
		    				HashtagEntity[] hashtags = status.getHashtagEntities();	
		    				
		    				/*if(hashtags.length > 0)
		    				{
		    					for(int i=0; i < hashtags.length ; i++)
		    					{
		    						HashtagEntity h = hashtags[i];				
		    						
		    					}
		    				}*/
		    	
		    				//******************** MY FORM **************************	 
		    				
		    				JSONArray hashtagsArray = new JSONArray();
		    				
		    				for(HashtagEntity hash : hashtags)
		    				{
		    					JSONObject hashObj= new JSONObject();
		    					hashObj.put("text", hash.getText());
		    					
		    					hashtagsArray.put(hashObj);
		    				}
		    				
		    				UserMentionEntity[] mentions=status.getUserMentionEntities();
		    				
		    				JSONArray mentionsArray = new JSONArray();
		    				
		    				for(UserMentionEntity mention : mentions)
		    					
		    				{
		    					JSONObject mentionObj= new JSONObject();
		    					mentionObj.put("name", mention.getScreenName());
		    					
		    					mentionsArray.put(mentionObj);
		    				}

		    				JSONObject tweetObject = new JSONObject();
		    				
					    	tweetObject.put("id", id);			
		    				tweetObject.put("username",username);
		    				tweetObject.put("retweets",retweets);
		    				tweetObject.put("followers",followers);
				
		    				tweetObject.put("lang", lang);
		    				tweetObject.put("createdat",createdat);
		    				tweetObject.put("millis", millis);
		    				tweetObject.put("text",text);
		    				tweetObject.put("lat", lat);
		    				tweetObject.put("lon",lon);
		    				tweetObject.put("country", country);
		    				
		    				tweetObject.put("hashtags", hashtagsArray);
		    				tweetObject.put("mentions", mentionsArray);
		    				
		    				String mytweet = tweetObject.toString();
		    				System.out.println("Counter: " + counter);
		    				System.out.println(mytweet);
		    				counter++;
			    	        try 
			    	        {
								bw.write(mytweet);
								bw.newLine();
							} 
			    	        catch (IOException e) 
			    	        {
								// TODO Auto-generated catch block
								e.printStackTrace();
			    	        }
	    				}
		    	        
	    				//******************** MY FORM *******************************	
	    				
	    				//******************** TWITTER FORM **************************	    
	
/*	    				String rawJSON = TwitterObjectFactory.getRawJSON(status);
		    	        System.out.println(rawJSON);
		    	        counter++;
		    	        System.out.println("Counter: " + counter);
		    	        try {
							bw.write(rawJSON);
							bw.newLine();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}*/
		    	      //******************** TWITTER FORM ****************************	
	    			}
	    		} 
				else 
	    		{
	    			System.out.println("Queue is now full, the following message is dropped: " + status.getText());
	    		}			
			}

			public void onTrackLimitationNotice(int arg0) {
				// TODO Auto-generated method stub
				
			}
		 
		 };
		 		 
		    ConfigurationBuilder cb = new ConfigurationBuilder();

		    cb.setDebugEnabled(true)
		      .setOAuthConsumerKey("6eHKCDId2GB72YsMrvgYeWVtU")
		      .setOAuthConsumerSecret("VDk8xgW3TElb5ErTLJcdImu3w5pVMs6Frm23SOeB8wYlPiI61a")
		      .setOAuthAccessToken("2462025612-2aOGrd3ZFZ1tv3Xfn4iCB3DggRoPPUFtRLxmyEL")
		      .setOAuthAccessTokenSecret("8pU9R2wNUmf7ghj1kL7cEqdnOBYKpA9AHxRQuaGcQaCCJ");
		    
		    cb.setJSONStoreEnabled(true);
		    cb.setIncludeEntitiesEnabled(true);
		 
		    twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
		    twitterStream.addListener(listener);
		    
		    FilterQuery filter = new FilterQuery();
		    double[][] locations = { { -180, -90 },
            { 180, 90 } };
		    filter.locations(locations);  
		    
		   	twitterStream.filter(filter);
		    
		    //twitterStream.sample();
		    
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("tweet"));
		
	}
}
